package com.edclol.apitest.tableapi;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Json;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;


public class TableTest4_KafkaPipeLine2 {
    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 2. 连接Kafka，读取数据
        tableEnv.connect(new Kafka()
                .version("0.11")
                .topic("json1")
                .property("zookeeper.connect", "node105:2181")
                .property("bootstrap.servers", "node105:9092")
                .startFromEarliest()
        )
                .withFormat(new Json())
                .withSchema(new Schema()
                        .field("data", DataTypes.ROW(
                                DataTypes.FIELD("active_day", DataTypes.STRING()),
                                DataTypes.FIELD("active_week", DataTypes.ARRAY(DataTypes.ROW(
                                        DataTypes.FIELD("rate", DataTypes.BIGINT()),
                                        DataTypes.FIELD("title", DataTypes.STRING())
                                ))),
                                DataTypes.FIELD("age", DataTypes.STRING()),
                                DataTypes.FIELD("city", DataTypes.STRING()),
                                DataTypes.FIELD("cons", DataTypes.STRING()),
                                DataTypes.FIELD("gender", DataTypes.MAP(DataTypes.STRING(), DataTypes.BIGINT())),
                                DataTypes.FIELD("has_feature", DataTypes.BOOLEAN()),
                                DataTypes.FIELD("province", DataTypes.STRING())
                        ))
                        .field("errCode", DataTypes.INT())
                        .field("author_id", DataTypes.STRING())

                )
                .createTemporaryTable("inputTable");
        Table inputTable = tableEnv.from("inputTable");
        inputTable.printSchema();
        tableEnv.toAppendStream(inputTable, Row.class).print();
      /*  // 3. 查询转换
        // 简单转换
        Table sensorTable = tableEnv.from("inputTable");
        Table resultTable = sensorTable.select("id, temp")
                .filter("id === 'sensor_6'");

        // 聚合统计
        Table aggTable = sensorTable.groupBy("id")
                .select("id, id.count as count, temp.avg as avgTemp");

        // 4. 建立kafka连接，输出到不同的topic下
        tableEnv.connect(new Kafka()
                .version("0.11")
                .topic("sinktest")
                .property("zookeeper.connect", "localhost:2181")
                .property("bootstrap.servers", "localhost:9092")
        )
                .withFormat(new Csv())
                .withSchema(new Schema()
                        .field("id", DataTypes.STRING())
//                        .field("timestamp", DataTypes.BIGINT())
                        .field("temp", DataTypes.DOUBLE())
                )
                .createTemporaryTable("outputTable");
                resultTable.insertInto("outputTable");
                */


        env.execute();
    }
}
